package com.keray.web.push;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.StrUtil;
import com.keray.web.push.store.SocketDataStore;
import com.keray.web.push.store.WorkerStore;
import com.keray.web.push.worker.SocketWorker;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import javax.websocket.Session;
import java.io.IOException;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

/**
 * @author by keray
 * date:2021/7/23 9:56 上午
 */
@Slf4j
@Service
public class WebSocket implements BeanPostProcessor {

    private static final AtomicInteger COUNT = new AtomicInteger(0);

    private static final ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(500, 1000, 10,
            TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(100000),
            r -> {
                Thread t = new Thread(r);
                t.setName("sys-thread-" + COUNT.getAndIncrement());
                return t;
            });

    @Resource
    private SocketDataStore socketDataStore;


    public void syncSend(String message, Session session) throws IOException {
        session.getBasicRemote().sendText(message);
    }

    public Future<Void> asyncSend(String message, Session session) {
        return session.getAsyncRemote().sendText(message);
    }

    public void syncSend(ClientMessageData data, Session session) throws IOException {
        SocketWorker worker = getWorker(data.getType());
        if (worker == null) {
            log.warn("发送消息没得对应的worker type={}", data.getType());
            return;
        }
        if (StrUtil.isBlank(data.getMessageId())) {
            data = worker.onSend(data);
        }
        if (data.getVersion() == 1) {
            syncSend("J:" + data.getMessage(), session);
        } else {
            syncSend(data.getMessage(), session);
        }
        worker.onSendOk(data);
    }


    public Future<Void> asyncSend(ClientMessageData data, Session session) {
        return threadPoolExecutor.submit(() -> {
            syncSend(data, session);
            return null;
        });
    }


    /**
     * <p>
     * <h3>作者 keray</h3>
     * <h3>时间： 2021/7/23 11:28 上午</h3>
     * 同步主动发送消息
     * </p>
     *
     * @param message
     * @param clientData
     * @return <p> {@link List <  ClientData >} </p>
     * 返回值表示发送失败的
     * @throws
     */
    public List<Session> syncSend(ClientMessageData message, ClientData clientData) {
        try {
            return asyncSend(message, clientData).get();
        } catch (InterruptedException | ExecutionException e) {
            log.error("send error", e);
        }
        return null;
    }

    /**
     * <p>
     * <h3>作者 keray</h3>
     * <h3>时间： 2021/7/23 11:28 上午</h3>
     * 异步主动发送消息
     * </p>
     *
     * @param message
     * @param clientData
     * @return <p> {@link List< ClientData>} </p>
     * 返回值表示发送失败的
     * @throws
     */
    public Future<List<Session>> asyncSend(ClientMessageData message, ClientData clientData) {
        return threadPoolExecutor.submit(() -> {
            List<Session> sessions = activeSession(clientData);
            if (CollUtil.isEmpty(sessions)) {
                return Collections.emptyList();
            }
            log.warn("找到发送session size={}，开始发送消息:{} {}", sessions.size(), message, clientData);
            List<Future<Session>> futures = sessions.stream().map(s -> threadPoolExecutor.submit(() -> {
                for (var i = 0; i < message.getRetry(); i++) {
                    try {
                        syncSend(message, s);
                        return null;
                    } catch (IOException ignore) {
                    }
                }
                return s;
            })).collect(Collectors.toList());
            return futures.stream().map(v -> {
                try {
                    return v.get();
                } catch (InterruptedException | ExecutionException e) {
                    log.error("发送消息失败：", e);
                    return null;
                }
            }).filter(Objects::nonNull).collect(Collectors.toList());
        });
    }


    private LinkedList<Session> activeSession(ClientData clientData) {
        if (StrUtil.isBlank(clientData.getService())) {
            throw new RuntimeException("no service");
        }
        LinkedList<Session> sessions;
        if (StrUtil.isNotBlank(clientData.getUid())) {
            return socketDataStore.serviceAndUidSession(clientData.getService(), clientData.getUid());
        } else if (StrUtil.isNotBlank(clientData.getGroup())) {
            sessions = socketDataStore.serviceAndGroupSession(clientData.getService(), clientData.getGroup());
        } else {
            sessions = socketDataStore.serviceSession(clientData.getService());
        }
        return sessions;
    }

    private SocketWorker getWorker(MessageType type) {
        return WorkerStore.MAP.get(type);
    }
}
